/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.scheduler;

import org.apache.log4j.Logger;

import java.io.IOException;
import java.net.Socket;
import java.util.Collection;

import jp.aist.gtrc.plus.scheduler.net.AcceptChecker;
import jp.aist.gtrc.plus.scheduler.net.GuardedRMIClientSocketFactory;
import jp.aist.gtrc.plus.scheduler.net.GuardedRMIServerSocketFactory;
import jp.aist.gtrc.plus.scheduler.node.NodeInfo;
import jp.aist.gtrc.plus.reserve.ReserveId;
import jp.aist.gtrc.plus.scheduler.specific.pbs.batch.PBSBatchServer;
import jp.aist.gtrc.plus.scheduler.specific.pbs.batch.actor.BatchActorAuthenticateUser;
import jp.aist.gtrc.plus.scheduler.specific.pbs.batch.actor.BatchActorDisconnect;
import jp.aist.gtrc.plus.scheduler.specific.pbs.batch.actor.PBSBatchActor;
import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSJobStatus;
import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSMainServer;
import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSNodeStatus;
import jp.aist.gtrc.plus.scheduler.specific.pbs.util.PBSScheduleStarter;
import jp.aist.gtrc.plus.scheduler.specific.pbs.util.PBSScheduleOrder;
import jp.aist.gtrc.plus.scheduler.specific.pbs.util.PBSSchedulerCommandType;
import jp.aist.gtrc.plus.scheduler.specific.pbs.util.PBSSchedulerOptions;
import jp.aist.gtrc.plus.scheduler.specific.pbs.util.PBSServerConfig;
import jp.aist.gtrc.plus.scheduler.specific.pbs.util.PBSServerConnector;

import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.status.MainServer;
import jp.aist.gtrc.plus.scheduler.status.NodeStatus;
import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;

import jp.aist.gtrc.plus.reserve.ReserveInfo;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocateRequest;


class PBSScheduler extends DefaultSelfScheduler {

        protected static final Logger logger = Logger.getLogger(PBSScheduler.class);

        protected PBSSchedulerOptions options;
        protected PBSServerConnector serverConnector;
        protected PBSServerConfig serverConf;
        protected PBSBatchServer authServer;
        protected AcceptChecker checker;

        private Collection<JobStatus> allJobStatus;

        PBSScheduler(SchedulerOptions options) throws Exception {

                super(options);

                init(options);

                startAuthenticateServer(serverConf.getAuthenticateServerPort());
                startReserveServer(options.getRMIPort());

        }

        private void init(SchedulerOptions scOpts) throws Exception {

                logger.debug("PBSScheduler init()");
                if (options == null) {
                        options = new PBSSchedulerOptions(scOpts.getArgs());
                        serverConf = new PBSServerConfig(options);
                        serverConnector = new PBSServerConnector(serverConf);
                        checker = new AcceptChecker();
                }

        }

        protected ScheduleStarter getScheduleStarter(SchedulerOptions scOpts) throws SchedulerException {

                try {
                        init(scOpts);
                } catch (Exception e) {
                        throw new SchedulerException("cannot initialize PBSScheduler");
                }

                return new PBSScheduleStarter(serverConf);

        }

        private void startAuthenticateServer(int authSrvPort) throws IOException{

                PBSBatchActor actor = new PBSBatchActor();
                actor.addActor(new BatchActorAuthenticateUser(checker));
                actor.addActor(new BatchActorDisconnect());
                authServer = new PBSBatchServer(authSrvPort, actor);
                Thread t = new Thread(authServer, "PBSAuthenticateServer");
                t.setDaemon(true);
                t.start();

        }

        private void stopAuthenticateServer() {

                if (authServer != null) {
                        authServer.finish();
                        authServer = null;
                }

        }

        protected void startReserveServer(int rmiPort) throws Exception {

                GuardedRMIClientSocketFactory csf
                        = new GuardedRMIClientSocketFactory(
                                        serverConf.getAuthenticateServerName(),
                                        serverConf.getAuthenticateServerPort());
                GuardedRMIServerSocketFactory ssf
                        = new GuardedRMIServerSocketFactory(checker);

                super.startReserveServer(csf, ssf);

        }

        protected MainServer getMainServer() throws Exception  {

                // get current pbs_server status and node information.
                // node information is needed for reserve node check.
                Socket s = serverConnector.getAuthorizedSocket();
                return (MainServer)new PBSMainServer(s);

        }

        /*
        protected ReserveManager getReserveManager(SchedulerOptions options) throws SchedulerException {

                NodeAllocator rsvNodeAllocator = getRsvNodeAllocator(options);
                if (rsvNodeAllocator == null) {
                        throw new SchedulerException("Invalid ReserveNodeAllocator");
                }

                return new PBSReserveManager(getReserveNodeManager(), starter,
                                options.getReserveFilePath(), rsvNodeAllocator,
                                new PBSQueueManager(serverConnector), options.getMaxExpiredReserves());

        }
        */

        protected void stopScheduler() {

                stopAuthenticateServer();
                super.stopScheduler();

        }

        protected boolean handleUnreservedJobs(MainServer srv, Collection<NodeInfo> rsvdNodes) {

                boolean reruned = false;

                for (NodeInfo node : rsvdNodes) {
                        // 'runningJobs' is all job names running on node with/without reserve.
                        NodeStatus s = node.getStatus();
                        assert(s instanceof PBSNodeStatus);
                        PBSNodeStatus ns = (PBSNodeStatus)s;
                        for (String jobId : ns.getRunningJobIDs()) {
                                JobStatus job = statusMgr.getJob(jobId);
                                if (job == null) {
                                        assert(false);
                                        continue;
                                }
                                ReserveId rsvId = job.getReserveId();
                                if (rsvId != null) {
                                        // 'job' is using reserve. OK to run.
                                        continue;
                                }

                                /*
                                 * 'job' is running without reserve.
                                 * We want to run reserved job on this 'node',
                                 * so rerurn un-reserved job here.
                                 * Reruned job will be killed by pbs_server, and restart
                                 * running at next scheduling cycle if possible.
                                 */
                                try {
                                        srv.rerunJob(job);
                                        reruned = true;
                                } catch (SchedulerException e) {
                                        logger.info("cannot rerun of " + job.getName());
                                }
                        }
                }


                return reruned;

        }

        protected void runSchedule(ScheduleOrder order) throws SchedulerException {
           long before = System.currentTimeMillis();

           //System.out.println("PBSScheduler is about to runSchedule");
           super.runSchedule(order);

           long after = System.currentTimeMillis();
//           System.out.println("PBSScheduler spent " + (after-before) + " msec. in scheduling");
           logger.debug("PBSScheduler spent " + (after-before) + " msec. in scheduling");

        }


        protected void schedule(ScheduleOrder order) throws SchedulerException{

                assert(jobMgr != null);
                assert(scheduler != null);

                //logger.debug(order.getClass().getName());

                if (order instanceof PBSScheduleOrder) {
                   PBSSchedulerCommandType cmd = ((PBSScheduleOrder) order).getSchedulerCommand();
                   logger.debug("PBSScheduler schedule() by command " + cmd);
                }

                Collection<JobStatus> allJobs = statusMgr.getJobs();

                //for (JobStatus jobStatus : allJobs) {
                //   PBSJobStatus pbsJobStatus = (PBSJobStatus) jobStatus;
                //   logger.debug(pbsJobStatus.getJobID() + " spent " + pbsJobStatus.getUsedCPUTime() + " sec.");  // resources_used.cput
                //}

                /*
                 * Select one job to run
                 */
                Collection<JobStatus> runnableJobs = jobMgr.getRunnableJobs(allJobs);
                if (runnableJobs != null) {
                        JobStatus job = scheduler.selectJob(runnableJobs);
                        runOneJob(order.getMainServer(), job);
                }

        }

        protected void runOneJob(MainServer srv, JobStatus job) throws SchedulerException {

                assert(nodeAllocator != null);

                /*
                 * Select exec-nodes to use
                 */
                ReserveId rsvId = job.getReserveId();
                ReserveInfo rsvInfo = rsvMgr.getReserveInfo(rsvId);
                Collection<NodeInfo> usableNodes, nodes;
                if (rsvInfo != null) {
                        usableNodes = rsvInfo.getReservedNodeInfos();
                        boolean mustRestart = handleUnreservedJobs(srv, usableNodes);
                        if (mustRestart) {
                                // some information was changed by handleUnreservedJobs(),
                                // don't call runJob(), restart scheduling
                                return;
                        }
                } else {
                        usableNodes = nodeMgr.getUnreservedNodes();
                }

                nodes = nodeAllocator.suitable(usableNodes, new NodeAllocateRequest(job));
                if (nodes != null) {
                        /*
                         * Run the job at nodes
                         */
                        srv.runJob(job, nodes);
                } else {
                        // no enough exec-nodes
                        // ex: reserve is correct, but too many jobs were submitted
                }

        }



}

